package com.study.flink.java.utils;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * Flink工具类
 * @author linys
 * @since 1.0.0
 */
public class FlinkUtilsDemo {

    // flink stream上下文
    private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    /**
     * 获取kafka_source
     * @param args
     * @param simpleStringSchema
     * @return
     */
    public static DataStream<String> createKafkaStream(String[] args, SimpleStringSchema simpleStringSchema) {

        String topic = args[0]; //wc10
        String gid = args[1]; //gid-wc10
        String brokerList = args[2]; //node02:9092

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", brokerList); //kafka的broker地址
        props.setProperty("group.id", gid);//指定组ID
        props.setProperty("auto.offset.reset", "earliest");//没有记录偏移量，第一次从最开始消费

        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(topic, simpleStringSchema, props);
        return env.addSource(kafkaSource);
    }

    public static StreamExecutionEnvironment getEnv() {
        return env;
    }


}
